#include <fermat/net/tcp_connection.h>
#include <fermat/event/channel.h>
#include <fermat/event/event_loop.h>
#include <fermat/net/tcp_socket.h>
#include <fermat/common/weak_call.h>
#include <fermat/common/logging.h>
#include <fermat/net/socket_stream.h>
#include <functional>
#include <cerrno>

namespace fermat{
namespace net  {

void defaultConnectionCallback(const TcpConnectionPtr& conn)
{
  	LOG_TRACE("{} -> {} {}", conn->localAddress().toStringHostPort(), 
                             conn->peerAddress().toStringHostPort(),
                            (conn->connected() ? " UP" : " DOWN"));
}

void defaultMessageCallback(const TcpConnectionPtr& conn,
                            PageBufferSliceQueuePtr buffer,
                            fermat::Timestamp receiveTime)

{
	(void)conn;
	(void)receiveTime;
  	buffer->clear();
}
typedef std::function<void (const Slice&)> connSendFunc;
struct ConnSendAdapter{
	explicit ConnSendAdapter(const connSendFunc& func, 
								const Slice&pice)
		:_func(func),
		 _data(pice.data(),pice.size())
	{
		assert(_func);
	}

	void doSend()
	{
		_func(Slice(_data.c_str(),_data.size()));
	}
	connSendFunc _func;
	std::string _data;
	

}; 

typedef std::function<void (fermat::PageBufferSliceQueuePtr)> connQueueSendFunc;
struct ConnSendQueueAdapter{
  explicit ConnSendQueueAdapter(const connQueueSendFunc& func, 
                fermat::PageBufferSliceQueuePtr &pice)
    :_func(func),
     _data(pice)
  {
    assert(_func);
  }

  void doSend()
  {
    _func(_data);
  }
  connQueueSendFunc            _func;
  PageBufferSliceQueuePtr _data;
}; 
TcpConnection::TcpConnection(fermat::event::EventLoopPtr loop,
                             const std::string& nameArg,
                             int sockfd,
                             const SocketAddress& localAddr,
                             const SocketAddress& peerAddr)
  : _lastactive(),
  	_timerOutSec(0),
  	_loop(loop),
    _name(nameArg),
    _state(kConnecting),
    _socket(new Socket(true)),
    _channel(new Channel(loop.get(), sockfd)),
    _inputBuffer(new PageBufferSliceQueue()),
    _outputBuffer(new PageBufferSliceQueue()),
    _localAddr(localAddr),
    _peerAddr(peerAddr),
    _highWaterMark(64*1024*1024)
{
  	_socket->setFd(sockfd);
  	_channel->setReadCallback(
  	std::bind(&TcpConnection::handleRead, this, std::placeholders::_1));
  	_channel->setWriteCallback(
  	std::bind(&TcpConnection::handleWrite, this));
  	_channel->setCloseCallback(
  	std::bind(&TcpConnection::handleClose, this));
  	_channel->setErrorCallback(
  	std::bind(&TcpConnection::handleError, this));
  	
  	LOG_DEBUG("TcpConnection::connect {} at fd= {} ", _name, sockfd);
  	_socket->setKeepAlive(true);
}

TcpConnection::~TcpConnection()
{
    LOG_DEBUG("TcpConnection::disconnect {} at fd= {} ", _name, _channel->fd());
    assert(_state == kDisconnected);
  
}

/*
void TcpConnection::enableTimerOut(int64_t secs) 
{
	if(_timerWheelHandler.valid()) {
		_timerWheelHandler.release();
	}
	_timerWheelHandler = _loop->runEverySec(secs,std::bind(&TcpConnection::onTimerOut,this));
}

void TcpConnection::disableTimerOut() 
{
	_timerWheelHandler.release();
}

void TcpConnection::onTimerOut()
{

	Time now = Time::now();
	Time diff = now - _lastactive;
	if(diff.toSeconds() >= _timerOutSec) {
		forceCloseInLoop();
	}
}
*/

bool TcpConnection::getTcpInfo(struct tcp_info* tcpi) const
{
    return _socket->getTcpInfo(tcpi);
}

std::string TcpConnection::getTcpInfoString() const
{
    char buf[4096];
    buf[0] = '\0';
    _socket->getTcpInfoString(buf, sizeof buf);
    return buf;
}

void TcpConnection::send(const void* data, int len)
{
  send(Slice(static_cast<const char*>(data), len));
}
void TcpConnection::send(PageBufferSliceQueuePtr ptr)
{
    if (_state == kConnected) {
        if (_loop->isInLoopThread()) {
            sendPageInLoop(ptr);
        } else  {
            connQueueSendFunc cb(std::bind(&TcpConnection::sendPageInLoop, this,ptr));
            ConnSendQueueAdapter ad(cb, ptr);
            _loop->runInLoop(
                std::bind(&ConnSendQueueAdapter::doSend,ad));
        }
    }
}
void TcpConnection::send(const Slice& message)
{
    if (_state == kConnected) {
        if (_loop->isInLoopThread()) {
            sendInLoop(message);
        } else  {
          	connSendFunc cb(std::bind(&TcpConnection::sendPiceInLoop, this,message));
          	ConnSendAdapter ad(cb, message);
            _loop->runInLoop(
                std::bind(&ConnSendAdapter::doSend,ad));
        }
    }
}

/*
void TcpConnection::send(IOBuffer* buf)
{
  	LOG_DEBUG("TcpConnection::send");
    if (_state == kConnected) {
        if (_loop->isInLoopThread()) {
            sendInLoop(buf->peek(), buf->readableBytes());
            buf->retrieveAll();
        } else {
            Slice message(buf->peek(),buf->readableBytes()); 
    	      connSendFunc cb = std::bind(&TcpConnection::sendPiceInLoop,this,std::placeholders::_1);
            ConnSendAdapter ad(cb, message);
    	      buf->retrieveAll();
            _loop->runInLoop(
                std::bind(&ConnSendAdapter::doSend,ad));
        }
    }
}
*/
void TcpConnection::sendInLoop(const Slice& message)
{
  	LOG_DEBUG("TcpConnection::sendInLoop");
    sendInLoop(message.data(), message.size());
}

void TcpConnection::sendPiceInLoop(const Slice& message)
{
  	LOG_DEBUG("TcpConnection::sendPiceInLoop");
    sendInLoop(message.data(), message.size());
}
void TcpConnection::sendPageInLoop(PageBufferSliceQueuePtr ptr)
{
    LOG_DEBUG("TcpConnection::sendInLoop ptr");
    _loop->assertInLoopThread();
    SocketStream<8> out(ptr);
    ssize_t nwrote = 0;
    size_t len = out.size();
    size_t remaining = out.size();
    bool faultError = false;
    if (_state == kDisconnected) {
        LOG_WARN("disconnected, give up writing");
        return;
    }
    _lastactive.update();
    // if no thing in output queue, try writing directly
    SocketStream<0> stream(_outputBuffer);
    if (!_channel->isWriting() && stream.size() == 0) {
        int err = 0;
        nwrote = out.writeToFd(_channel->fd(), 0, err);
        if (nwrote >= 0) {
            remaining = len - nwrote;
            if (remaining == 0 && _writeCompleteCallback) {
                _loop->queueInLoop(std::bind(_writeCompleteCallback,shared_from_this()));
            }
        } else {
            nwrote = 0;
            if (errno != EWOULDBLOCK) {
                LOG_ERROR("TcpConnection::sendInLoop error {}",  errno);
                if (errno == EPIPE || errno == ECONNRESET) {
                  faultError = true;
                }
            }
        }
    }

    assert(remaining <= len);
    if (!faultError && remaining > 0) {
        size_t oldLen = stream.size();
        if (oldLen + remaining >= _highWaterMark
            && oldLen < _highWaterMark
            && _highWaterMarkCallback) {
            _loop->queueInLoop(std::bind(_highWaterMarkCallback, 
            shared_from_this(), oldLen + remaining));
        }

        while(ptr->size() > 0) {
            fermat::PageBufferSlicePtr pptr = ptr->front();
            _outputBuffer->push_back(pptr);
            ptr->pop_front();
        }
        if (!_channel->isWriting()) {
            _channel->enableWriting();
        }
    }
}
void TcpConnection::sendInLoop(const void* data, size_t len)
{
  	LOG_DEBUG("TcpConnection::sendInLoop");
    _loop->assertInLoopThread();
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;
    if (_state == kDisconnected) {
        LOG_WARN("disconnected, give up writing");
        return;
    }
  	_lastactive.update();
    // if no thing in output queue, try writing directly
    SocketStream<0> stream(_outputBuffer);
    if (!_channel->isWriting() && stream.size() == 0) {
        nwrote = ::write(_channel->fd(), data, len);
        if (nwrote >= 0) {
            remaining = len - nwrote;
            if (remaining == 0 && _writeCompleteCallback) {
              _loop->queueInLoop(std::bind(_writeCompleteCallback,shared_from_this()));
            }
        } else {
            nwrote = 0;
            if (errno != EWOULDBLOCK) {
                LOG_ERROR("TcpConnection::sendInLoop error {}",  errno);
                if (errno == EPIPE || errno == ECONNRESET) {
                  faultError = true;
                }
            }
        }
    }

    assert(remaining <= len);
    if (!faultError && remaining > 0) {
        size_t oldLen = stream.size();
        if (oldLen + remaining >= _highWaterMark
            && oldLen < _highWaterMark
            && _highWaterMarkCallback) {
            _loop->queueInLoop(std::bind(_highWaterMarkCallback, 
  	  	    shared_from_this(), oldLen + remaining));
        }
        const char *p = static_cast<const char*>(data);
        while(remaining > 0) {
          size_t avial;
          fermat::PageBufferSlicePtr ptr(new fermat::PageBufferSlice());
          if(remaining < ptr->writeAbleSize()) {
              avial = remaining;
          } else {
              avial = ptr->writeAbleSize();
          }
          ptr->append(p, avial);
          remaining -= avial;
          _outputBuffer->push_back(ptr);
        }
        if (!_channel->isWriting()) {
            _channel->enableWriting();
        }
    }
}

void TcpConnection::shutdown()
{
  	LOG_DEBUG("TcpConnection::shutdown");
    // FIXME: use compare and swap
    if (_state == kConnected) {
        setState(kDisconnecting);
        // FIXME: shared_from_this()?
        _loop->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
  	    _loop->wakeup();
    }
}

void TcpConnection::shutdownInLoop()
{
  	LOG_DEBUG("TcpConnection::shutdownInLoop");
    _loop->assertInLoopThread();
    if (!_channel->isWriting()) {
      // we are not writing
      _socket->shutdownWR();
    }
}

void TcpConnection::forceClose()
{
    // FIXME: use compare and swap
    LOG_DEBUG("TcpConnection::forceClose");
    if (_state == kConnected || _state == kDisconnecting) {
      setState(kDisconnecting);
      _loop->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, 
  			shared_from_this()));
    }
}

void TcpConnection::forceCloseWithDelay(int64_t seconds)
{
  	LOG_DEBUG("TcpConnection::forceCloseWithDelay");
    if (_state == kConnected || _state == kDisconnecting) {
        setState(kDisconnecting);

        _loop->runAfter(
            seconds*1000000,
            makeWeakCallback(shared_from_this(),
                             &TcpConnection::forceClose)); 
    }
}

void TcpConnection::forceCloseInLoop()
{
  	LOG_DEBUG("TcpConnection::forceCloseInLoop");

    _loop->assertInLoopThread();
    if (_state == kConnected || _state == kDisconnecting) {
      // as if we received 0 byte in handleRead();
        _socket->shutdownAll();
        handleClose();
    }
}

void TcpConnection::setTcpNoDelay(bool on)
{
    _socket->setTcpNoDelay(on);
}

void TcpConnection::setTcpLiner(bool on, int sec)
{
    _socket->setSoLinger(on, sec);  
}
void TcpConnection::connectEstablished()
{
    LOG_DEBUG("TcpConnection::connectEstablished");
    _loop->assertInLoopThread();
    assert(_state == kConnecting);
    setState(kConnected);
    _channel->tie(shared_from_this());
    _channel->enableReading();

    _connectionCallback(shared_from_this());
}

void TcpConnection::connectDestroyed()
{
		
    _loop->assertInLoopThread();
    if (_state == kConnected) {
      setState(kDisconnected);
      _channel->disableAll();

      _connectionCallback(shared_from_this());
    }
    _channel->remove();
    LOG_DEBUG("TcpConnection::connectDestroyed");
}

void TcpConnection::handleRead(fermat::Timestamp receiveTime)
{
    LOG_DEBUG("TcpConnection::handleRead");

    _loop->assertInLoopThread();
    SocketStream<8> stream(_inputBuffer);
    int savedErrno = 0;
      _lastactive.update();
    ssize_t n = stream.readFromFd(_channel->fd(), savedErrno);
    if (n > 0) {
        _messageCallback(shared_from_this(), _inputBuffer, receiveTime);
    } else if (n == 0) {
        handleClose();
    } else {
      errno = savedErrno;
        LOG_ERROR("TcpConnection::handleRead");
        handleError();
    }
   
}

void TcpConnection::handleWrite()
{
    LOG_DEBUG("TcpConnection::handleWrite");
    _loop->assertInLoopThread();
    if (_channel->isWriting()) {
      _lastactive.update();
      SocketStream<8> stream(_outputBuffer);
      int serr = 0;
      ssize_t n = stream.writeToFd(_channel->fd(), 0, serr);
 
      if (n > 0) {
          if (stream.size() == 0) {
              _channel->disableWriting();
              if (_writeCompleteCallback) {
                  _loop->queueInLoop(std::bind(_writeCompleteCallback, 
                      shared_from_this()));
              }
              if (_state == kDisconnecting) {
                  shutdownInLoop();
              }
          }
      } else {
        if(errno != EAGAIN ) 
        handleClose();
        
      }
    } else {
      LOG_TRACE("Connection fd = {} is down, no more writing", _channel->fd());
    }
}

void TcpConnection::handleClose()
{
    LOG_DEBUG("TcpConnection::handleClose");
    _loop->assertInLoopThread();
    LOG_TRACE("fd = {} state = {}", _channel->fd(), (int)_state);
    assert(_state == kConnected || _state == kDisconnecting);
    // we don't close fd, leave it to dtor, so we can find leaks easily.
    setState(kDisconnected);
    _channel->disableAll();

    TcpConnectionPtr guardThis(shared_from_this());
    _connectionCallback(guardThis);
    // must be the last line
    _closeCallback(guardThis);
}

void TcpConnection::handleError()
{
    int err = _socket->getSoError();
    LOG_ERROR("TcpConnection::handleError {} - SO_ERROR = {}", _name, err);
}

}
}
